package com.atguigu.flink.datastreamapi.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;

/**
 * Created by Smexy on 2023/4/4
 *
 *  1.在依赖中添加 kafka connector
 *  2.flink程序在kafka的使用场景下的角色
 *          读kafka： 消费者。 可以使用消费者的所有的参数配置。
 *                      ConsumerConfig: 保存了所有消费者配置相关的key
 *          写kafka： 生产者
 *                      ProducerConfig: 保存了生产者的配置的key
 *
 *  -----------------------------------------
 *       <OUT> KafkaSourceBuilder<OUT> builder():
 *          以上是一个泛型方法，调用泛型方法时，必须声明泛型的类型。
 *
 *              <T>builder()
 *
 */
public class Demo3_KafkaSource
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(5000);

        //构造KafkaSource   使用建造者模式构造
        KafkaSource<String> kafkaSource = KafkaSource
            .<String>builder()
            .setBootstrapServers("hadoop102:9092")  //如何连接集群
            .setGroupId("atguigutest") //设置组名
            .setTopics("topicB")   //设置读取哪个主题
            /*
            如果当前组是第一次消费，
               lateset： 默认，从当前topic的最后位置读取
               earliest: 从当前topic的最早读取。

               如果当前组已经消费过了，从上次消费的位置往后读。

               OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST):
                       从当前组上次提交的位置读取，如果是一个新的组，从主题的最早位置读取

               策略不写，默认就是latest
             */
            .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))  //设置从哪个位置读取
            /*
                生产：  数据---->ProducerRecord ---->序列化为 byte[] ----->broker-->存储到文件
                消费:   broker-->存储到文件---->byte[] ----->反序列化为 ConsumerRecord

                ProducerRecord和ConsumerRecord都是KEY-VALUE结构。key只有一个作用，就是在生产的时候分区。
                    相同的key的数据会自动放入同一个partition.
                    数据都是在value中，因此消费者消费时，只需要消费value!

                    .setValueOnlyDeserializer(): 只消费value，只需要设置value的反序列化器。 99%
                    .setDeserializer(): 消费key和value.设置key和value如何反序列化。
             */
            .setValueOnlyDeserializer(new SimpleStringSchema())
            //设置其他的消费者参数
            .setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
            .setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "500")
            .build();


        //把source添加到环境中
        env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(),"kafkaSource")
           .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
